package com.atguigu.flink.chapter10;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


import static org.apache.flink.table.api.Expressions.*;

/*
将动态表转换为流
    	仅追加（Append-only）流：动态表中新增的每一行
    	撤回（Retract）流：add消息；DELETE删除
    	更新插入（Upsert）流：INSERT插入操作和UPDATE更新操作、DELETE删除操作

但是使用调用toChangelogStream()就可以撤回流
 */
public class TableApiDemo2 {
    public static void main(String[] args) throws Exception {
        //表的执行
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("s1", 1000L, 10),
                new WaterSensor("s2", 2000L, 20),
                new WaterSensor("s1", 3000L, 30),
                new WaterSensor("s1", 4000L, 40),
                new WaterSensor("s2", 5000L, 50)
        );

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream);

        //api的方式：select id,sum(vc) as vc_sum from t group by id;
        Table result = table.groupBy($("id"))
                .aggregate($("vc").sum().as("vc_sum"))
                .select($("id"), $("vc_sum"));

        DataStream<Row> s1 = tEnv.toChangelogStream(result);

        //row.getKind().name(): 输出 -> INSERT/UPDATE_AFTER/UPDATE_BEFORE/DELETE
        s1
            .filter(row -> {
            String name = row.getKind().name();
            return  "INSERT".equals(name) || "UPDATE_AFTER".equals(name);
            })
            //将里面的数据封装成pojo类型
            .map(new MapFunction<Row, WaterSensor>() {
                @Override
                public WaterSensor map(Row row) throws Exception {
                    String id = row.getFieldAs("id");
                    Integer vcSum = row.getFieldAs("vc_sum");
                    return new WaterSensor(id,1L,vcSum);
                }
            })
            .print();


        env.execute();
    }
}


//public class TableApiDemo2 {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s1", 3000L, 30),
//                new WaterSensor("s1", 4000L, 40),
//                new WaterSensor("s2", 5000L, 50)
//        );
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        Table table = tEnv.fromDataStream(stream);
//
//        //api的方式：select id,sum(vc) as vc_sum from t group by id;
//        Table result = table.groupBy($("id"))
//                .aggregate($("vc").sum().as("vc_sum"))
//                .select($("id"), $("vc_sum"));
//
//        DataStream<Row> s1 = tEnv.toChangelogStream(result);
//
//        //row.getKind().name(); 输出的是 ： INSERT/UPDATE_BEFORE/UPDATE_AFTER/DELETE
//        s1.filter(row -> {
//            String name = row.getKind().name();
//            return "INSERT".equals(name) || "UPDATE_AFTER".equals(name);
//        })
//                //把filter后的数据封装成pojo类型
//                .map(new MapFunction<Row, WaterSensor>() {
//                    @Override
//                    public WaterSensor map(Row row) throws Exception {
//                        String id = row.getFieldAs("id");
//                        Integer vcsum = row.getFieldAs("vc_sum");
//                        return new WaterSensor(id,1L,vcsum);
//                    }
//                })
//                .print();
//
//        env.execute();
//    }
//}

//
//public class TableApiDemo2 {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s1", 3000L, 30),
//                new WaterSensor("s1", 4000L, 40),
//                new WaterSensor("s2", 5000L, 50)
//        );
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        Table table = tEnv.fromDataStream(stream);
//
//        //api的方式：select id,sum(vc) as vc_sum from t group by id
//        Table result = table.groupBy($("id"))
//                .aggregate($("vc").sum().as("vc_sum"))
//                .select($("id"), $("vc_sum"));
//
//        DataStream<Row> s1 = tEnv.toChangelogStream(result);
//
//        s1.filter(row -> {
//            String name = row.getKind().name();
//            return "INSERT".equals(name)  || "update_after".equals(name);
//        })
//                        .map(new MapFunction<Row, WaterSensor>() {
//                            @Override
//                            public WaterSensor map(Row row) throws Exception {
//                                String id = row.getFieldAs("id");
//                                Integer vcsum = row.getFieldAs("vc_sum");
//                                return new WaterSensor(id,1L,vcsum);
//                            }
//                        })
//                                .print();
//
//        env.execute();
//    }
//}